package com.lagou.bak;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

public class CEPActiveUser {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        DataStreamSource<ActiveUserBean> data = env.fromElements(
                new ActiveUserBean("100XX", 0.0D, 1597905234000L),
                new ActiveUserBean("100XX", 100.0D, 1597905235000L),
                new ActiveUserBean("100XX", 200.0D, 1597905236000L),
                new ActiveUserBean("100XX", 300.0D, 1597905237000L),
                new ActiveUserBean("100XX", 400.0D, 1597905238000L),
                new ActiveUserBean("100XX", 500.0D, 1597905239000L),
                new ActiveUserBean("101XX", 0.0D, 1597905240000L),
                new ActiveUserBean("101XX", 100.0D, 1597905241000L)
        );
        SingleOutputStreamOperator<ActiveUserBean> watermark = data.assignTimestampsAndWatermarks(new WatermarkStrategy<ActiveUserBean>() {
            @Override
            public WatermarkGenerator<ActiveUserBean> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<ActiveUserBean>() {
                    long maxTimeStamp = Long.MIN_VALUE;

                    @Override
                    public void onEvent(ActiveUserBean event, long eventTimestamp, WatermarkOutput output) {
                        maxTimeStamp = Math.max(maxTimeStamp, event.getTs());
                    }

                    long maxOutOfOrderness = 500l;

                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {
                        output.emitWatermark(new Watermark(maxTimeStamp - maxOutOfOrderness));
                    }
                };
            }
        }.withTimestampAssigner(((element, recordTimestamp) -> element.getTs())));
        KeyedStream<ActiveUserBean, String> keyed = watermark.keyBy(value -> value.getUid());
        Pattern<ActiveUserBean, ActiveUserBean> pattern = Pattern.<ActiveUserBean>begin("start").where(new SimpleCondition<ActiveUserBean>() {
            @Override
            public boolean filter(ActiveUserBean value) throws Exception {
                return value.getMoney() > 0;
            }
        }).timesOrMore(5);

        PatternStream<ActiveUserBean> patternStream = CEP.pattern(keyed, pattern);
        SingleOutputStreamOperator<ActiveUserBean> process = patternStream.process(new PatternProcessFunction<ActiveUserBean, ActiveUserBean>() {
            @Override
            public void processMatch(Map<String, List<ActiveUserBean>> match, Context ctx, Collector<ActiveUserBean> out) throws Exception {
                System.out.println(match);
            }
        });
        process.print();
        env.execute();

    }
}
